#include <glog/logging.h>

#include "index/attribute_index/attribute_filter/attribute_filter.h"

namespace hawking {
namespace indexlib {

bool AttributeFilter::DoFilter() const {
    if (is_where_) {
        return !DoFilter_();
    }

    return DoFilter_();
}

bool AttributeFilter::EqualCompare_() const {
    switch (field_type_) {
    case util::COLUMN_INT8:
    case util::COLUMN_INT16:
    case util::COLUMN_INT32:
    case util::COLUMN_INT64:
        return filter_field_value_.int_value() == index_field_value_.int_value();
    case util::COLUMN_FLOAT:
        return static_cast<float>(filter_field_value_.double_value())
            == static_cast<float>(index_field_value_.double_value());
    case util::COLUMN_DOUBLE:
        return filter_field_value_.double_value() == index_field_value_.double_value();
    case util::COLUMN_STRING:
        return filter_field_value_.bytes_value() == index_field_value_.bytes_value();
    case util::COLUMN_INT8_LIST:
    case util::COLUMN_INT16_LIST:
    case util::COLUMN_INT32_LIST:
    case util::COLUMN_INT64_LIST:
        {
            if (filter_field_value_.int_values().values_size() !=
                index_field_value_.int_values().values_size()) {
                return false;
            }

            for (int32_t idx = 0;
                 idx < filter_field_value_.int_values().values_size();
                 ++idx) {
                if (filter_field_value_.int_values().values(idx) !=
                    index_field_value_.int_values().values(idx)) {
                    return false;
                }
            }

            return true;
        }
    case util::COLUMN_FLOAT_LIST:
        {
            if (filter_field_value_.double_values().values_size() !=
                index_field_value_.double_values().values_size()) {
                return false;
            }

            for (int32_t idx = 0;
                 idx < filter_field_value_.double_values().values_size();
                 ++idx) {
                if (static_cast<float>(filter_field_value_.double_values().values(idx))
                    != static_cast<float>(index_field_value_.double_values().values(idx))) {
                    return false;
                }
            }

            return true;
        }
    case util::COLUMN_DOUBLE_LIST:
        {
            if (filter_field_value_.double_values().values_size() !=
                index_field_value_.double_values().values_size()) {
                return false;
            }

            for (int32_t idx = 0;
                 idx < filter_field_value_.double_values().values_size();
                 ++idx) {
                if (filter_field_value_.double_values().values(idx) !=
                    index_field_value_.double_values().values(idx)) {
                    return false;
                }
            }

            return true;
        }
    case util::COLUMN_STRING_LIST:
        {
            if (filter_field_value_.bytes_values().values_size() !=
                index_field_value_.bytes_values().values_size()) {
                return false;
            }

            for (int32_t idx = 0;
                 idx < filter_field_value_.bytes_values().values_size();
                 ++idx) {
                if (filter_field_value_.bytes_values().values(idx) !=
                    index_field_value_.bytes_values().values(idx)) {
                    return false;
                }
            }

            return true;
        }
    default:
        break;
    }

    return false;
}

bool AttributeFilter::GreaterCompare_() const {
    switch (field_type_) {
    case util::COLUMN_INT8:
    case util::COLUMN_INT16:
    case util::COLUMN_INT32:
    case util::COLUMN_INT64:
        return index_field_value_.int_value() > filter_field_value_.int_value();
    case util::COLUMN_FLOAT:
        return static_cast<float>(index_field_value_.double_value())
            > static_cast<float>(filter_field_value_.double_value());
    case util::COLUMN_DOUBLE:
        return index_field_value_.double_value() > filter_field_value_.double_value();
    case util::COLUMN_STRING:
        return index_field_value_.bytes_value() > filter_field_value_.bytes_value();
    default:
        break;
    }

    return false;
}

bool AttributeFilter::LowerCompare_() const {
    switch (field_type_) {
    case util::COLUMN_INT8:
    case util::COLUMN_INT16:
    case util::COLUMN_INT32:
    case util::COLUMN_INT64:
        return index_field_value_.int_value() < filter_field_value_.int_value();
    case util::COLUMN_FLOAT:
        return static_cast<float>(index_field_value_.double_value())
            < static_cast<float>(filter_field_value_.double_value());
    case util::COLUMN_DOUBLE:
        return index_field_value_.double_value() < filter_field_value_.double_value();
    case util::COLUMN_STRING:
        return index_field_value_.bytes_value() < filter_field_value_.bytes_value();
    default:
        break;
    }

    return false;
}

bool AttributeFilter::InCompare_() const {
    switch (field_type_) {
    case util::COLUMN_INT8:
    case util::COLUMN_INT16:
    case util::COLUMN_INT32:
    case util::COLUMN_INT64:
        return std::binary_search(
            filter_field_value_.int_values().values().begin(),
            filter_field_value_.int_values().values().end(),
            index_field_value_.int_value());
    case util::COLUMN_FLOAT:
    case util::COLUMN_DOUBLE:
        return std::binary_search(
            filter_field_value_.double_values().values().begin(),
            filter_field_value_.double_values().values().end(),
            index_field_value_.double_value());
    case util::COLUMN_STRING:
        return std::binary_search(
            filter_field_value_.bytes_values().values().begin(),
            filter_field_value_.bytes_values().values().end(),
            index_field_value_.bytes_value());
    default:
        break;
    }

    return false;
}

bool AttributeFilter::ContainCompare_() const {
    switch (field_type_) {
    case util::COLUMN_INT8_LIST:
    case util::COLUMN_INT16_LIST:
    case util::COLUMN_INT32_LIST:
    case util::COLUMN_INT64_LIST:
        return std::binary_search(
            index_field_value_.int_values().values().begin(),
            index_field_value_.int_values().values().end(),
            filter_field_value_.int_value());
    case util::COLUMN_FLOAT_LIST:
        // TODO optim
        for (int32_t idx = 0;
             idx < index_field_value_.double_values().values_size();
             ++idx) {
            if (static_cast<float>(index_field_value_.double_values().values(idx))
                == filter_field_value_.double_value()) {
                return true;
            }
        }

        break;
    case util::COLUMN_DOUBLE_LIST:
        return std::binary_search(
            index_field_value_.double_values().values().begin(),
            index_field_value_.double_values().values().end(),
            filter_field_value_.double_value());
    case util::COLUMN_STRING_LIST:
        return std::binary_search(
            index_field_value_.bytes_values().values().begin(),
            index_field_value_.bytes_values().values().end(),
            filter_field_value_.bytes_value());
    default:
        break;
    }

    return false;
}

bool AttributeFilter::IntersectCompare_() const {
     switch (field_type_) {
    case util::COLUMN_INT8_LIST:
    case util::COLUMN_INT16_LIST:
    case util::COLUMN_INT32_LIST:
    case util::COLUMN_INT64_LIST:
        {
            for (const auto elem : index_field_value_.int_values().values()) {
                if (std::binary_search(
                        filter_field_value_.int_values().values().begin(),
                        filter_field_value_.int_values().values().end(),
                        elem)) {
                    return true;
                }
            }

            return false;
        }
    case util::COLUMN_FLOAT_LIST:
    case util::COLUMN_DOUBLE_LIST:
        {
            for (const auto elem : index_field_value_.double_values().values()) {
                if (std::binary_search(
                        filter_field_value_.double_values().values().begin(),
                        filter_field_value_.double_values().values().end(),
                        elem)) {
                    return true;
                }
            }

            return false;
        }
    case util::COLUMN_STRING_LIST:
        {
            for (const auto& elem : index_field_value_.bytes_values().values()) {
                if (std::binary_search(
                        filter_field_value_.bytes_values().values().begin(),
                        filter_field_value_.bytes_values().values().end(),
                        elem)) {
                    return true;
                }
            }

            return false;
        }
    default:
        break;
    }

    return false;
}

bool AttributeFilter::DoFilter_() const {
    switch (op_) {
        case util::ExpressionOp::EQ:
            return EqualCompare_();
        case util::ExpressionOp::NE:
            return !EqualCompare_();
        case util::ExpressionOp::GT:
            return GreaterCompare_();
        case util::ExpressionOp::GE:
            return !LowerCompare_();
        case util::ExpressionOp::LT:
            return LowerCompare_();
        case util::ExpressionOp::LE:
            return !GreaterCompare_();
        case util::ExpressionOp::IN:
            return InCompare_();
        case util::ExpressionOp::NI:
            return !InCompare_();
        case util::ExpressionOp::CONTAIN:
            return ContainCompare_();
        case util::ExpressionOp::INTERSECT:
            return IntersectCompare_();
        default:
            LOG(ERROR) << "Unknown action " << op_;
    }

    return false;
}

}
}
